package com.roy.minimq.remoting.domain.nettyclient;

import com.roy.minimq.remoting.domain.entity.RemotingClient;
import com.roy.minimq.remoting.domain.entity.RemotingCommand;
import com.roy.minimq.remoting.domain.entity.ResponseFuture;
import com.roy.minimq.remoting.domain.netty.NettyDecoder;
import com.roy.minimq.remoting.domain.netty.NettyEncoder;
import com.roy.minimq.remoting.domain.netty.NettyRemotingAbstract;
import com.roy.minimq.remoting.exception.RemotingException;
import com.roy.minimq.remoting.service.ChannelEventListener;
import com.roy.minimq.remoting.service.InvokeCallback;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.DefaultEventExecutorGroup;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author ：楼兰
 * @date ：Created in 2021/8/9
 * @description:
 **/

public class NettyRemotingClient extends NettyRemotingAbstract implements RemotingClient {

    private Logger logger = LoggerFactory.getLogger(this.getClass());

    private Bootstrap bootstrap;
    private EventLoopGroup workerGroup;
    private NettyClientConfig nettyClientConfig;
    private DefaultEventExecutorGroup defaultEventExecutorGroup;
    private AtomicReference<List<String>> namesrvAddrList = new AtomicReference<>();
    private final AtomicReference<String> namesrvAddrChoosed = new AtomicReference<String>();
    private final ConcurrentMap<String /* addr */, ChannelFuture> channelTables = new ConcurrentHashMap<>();
    private Lock lock = new ReentrantLock();

    public ConcurrentHashMap<Integer, ResponseFuture> getResponseTable() {
        return responseTable;
    }

    private ChannelEventListener channelEventListener;

    public NettyRemotingClient(NettyClientConfig nettyClientConfig) {
        this.nettyClientConfig = nettyClientConfig;
        this.bootstrap = new Bootstrap();
        this.workerGroup = new NioEventLoopGroup(1, new ThreadFactory() {
            private AtomicInteger threadIndex = new AtomicInteger(0);

            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, String.format("NettyClientSelector_%d", this.threadIndex.incrementAndGet()));
            }
        });

        this.onewaySemaphore = new Semaphore(nettyClientConfig.getClientOnewaySemaphoreValue());
        this.asyncSemaphore = new Semaphore(nettyClientConfig.getClientAsyncSemaphoreValue());
        int publicThreadNums = Runtime.getRuntime().availableProcessors();
        if (publicThreadNums <= 0) {
            publicThreadNums = 4;
        }

        this.publicExecutor = Executors.newFixedThreadPool(publicThreadNums, new ThreadFactory() {
            private AtomicInteger threadIndex = new AtomicInteger(0);

            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "NettyClientPublicExecutor_" + this.threadIndex.incrementAndGet());
            }
        });
    }


    @Override
    public void start() {
        this.defaultEventExecutorGroup = new DefaultEventExecutorGroup(nettyClientConfig.getClientWorkThreads(), new ThreadFactory() {
            private AtomicInteger threadIndex = new AtomicInteger(0);

            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "NettyClientWorkerThread_" + this.threadIndex.incrementAndGet());
            }
        });

        final NettyClientHandler nettyClientHandler = new NettyClientHandler(this);
        final NettyClientConnectionManageHandler nettyClientConnectionManageHandler = new NettyClientConnectionManageHandler(this);
        this.bootstrap.group(this.workerGroup)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.SO_KEEPALIVE, false)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, nettyClientConfig.getConnectTimeoutMillis())
                .option(ChannelOption.SO_SNDBUF, nettyClientConfig.getClientSocketSndBufSize())
                .option(ChannelOption.SO_RCVBUF, nettyClientConfig.getClientSocketRcvBufSize())
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        final ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(defaultEventExecutorGroup,
                                new NettyDecoder(),
                                new NettyEncoder(),
                                new IdleStateHandler(0,0, nettyClientConfig.getClientChannelMaxIdleTimeSeconds()),
                                nettyClientConnectionManageHandler,
                                nettyClientHandler);
                    }
                });
        //测试用
//        try{
//            final ChannelFuture future = bootstrap.connect("127.0.0.1", 8888).sync();
//            logger.info("client is ready ");
//            future.channel().closeFuture().sync();
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }


    }

    @Override
    public void shutdown() {
        if (null != workerGroup) {
            workerGroup.shutdownGracefully();
        }
        if (null != this.defaultEventExecutorGroup) {
            this.defaultEventExecutorGroup.shutdownGracefully();
        }
        if (null != this.publicExecutor) {
            this.publicExecutor.shutdown();
        }
    }

    @Override
    public List<String> getNameServerAddressList() {
        return this.namesrvAddrList.get();
    }

    @Override
    public RemotingCommand invokeSync(String addr, RemotingCommand request, long timeoutMillis) throws RemotingException {
        //根据addr找到channel.
        Channel channel = this.getOrCreateChannel(addr);
        //拿到channel直接发送数据。
        if (null != channel && channel.isActive()) {
            try {
                RemotingCommand response = this.doInvokeSync(channel, request, timeoutMillis);
                return response;
                //同步等待的异常暂不处理。
            } catch (InterruptedException e) {
            }
        } else {
            this.closeChannel(addr, channel);
            throw new RemotingException("获取连接失败，连接目标：" + addr);
        }
        return null;
    }

    @Override
    public void invokeOneway(String addr, RemotingCommand request, long timeoutMillis) throws RemotingException {
        Channel channel = this.getOrCreateChannel(addr);
        if (null != channel && channel.isActive()) {
            try {
                this.doInvokeOneway(channel, request, timeoutMillis);
            } catch (InterruptedException e) {

            }
        } else {
            this.closeChannel(addr, channel);
            throw new RemotingException("获取连接失败，连接目标：" + addr);
        }
    }


    /**
     * 异步调用，在单向服务调用的基础上增加失败时的回调方法。
     *
     * @param addr
     * @param request
     * @param timeoutMillis
     * @param invokeCallback
     * @throws RemotingException
     */
    @Override
    public void invokeAsync(String addr, RemotingCommand request, long timeoutMillis, InvokeCallback invokeCallback) throws RemotingException {
        Channel channel = this.getOrCreateChannel(addr);
        if (null != channel && channel.isActive()) {
            try {
                this.doInvokeAsync(channel, request, timeoutMillis, invokeCallback);
            } catch (InterruptedException e) {
            }
        } else {
            this.closeChannel(addr, channel);
            throw new RemotingException("获取连接失败，连接目标：" + addr);
        }
    }

    private Channel getOrCreateChannel(String addr) {
        ChannelFuture channelFuture = this.channelTables.get(addr);
        if (null != channelFuture && channelFuture.channel().isActive()) {
            return channelFuture.channel();
        }
        //没有缓存就创建一个新的
        try {
            if (lock.tryLock(3000, TimeUnit.MILLISECONDS)) {
                try {
                    boolean createNewConnection;
                    channelFuture = this.channelTables.get(addr);
                    if (channelFuture != null) {

                        if (channelFuture.channel().isActive()) {
                            return channelFuture.channel();
                        } else if (!channelFuture.isDone()) {
                            createNewConnection = false;
                        } else {
                            this.channelTables.remove(addr);
                            createNewConnection = true;
                        }
                    } else {
                        createNewConnection = true;
                    }
                    if (createNewConnection) {
                        int split = addr.lastIndexOf(":");
                        String host = addr.substring(0, split);
                        String port = addr.substring(split + 1);
                        InetSocketAddress isa = new InetSocketAddress(host, Integer.parseInt(port));

                        channelFuture = this.bootstrap.connect(isa);
                        this.channelTables.put(addr, channelFuture);
                    }
                } catch (Exception e) {
                    logger.warn("create channel to addr:{} failed", addr, e);
                } finally {
                    lock.unlock();
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        if (channelFuture.awaitUninterruptibly(this.nettyClientConfig.getConnectTimeoutMillis())) {
            //等待连接创建完成
            if (channelFuture.channel().isActive()) {
                return channelFuture.channel();
            } else {
                logger.warn("create channel to addr:{} failed", addr);
            }
        } else {
            //创建连接超时
            logger.warn("create channel to addr:{} time out", addr);
        }

        return null;
    }


    //关闭与指定目标之间的连接
    public void closeChannel(String addr, Channel channel) {
        if (null == channel) {
            logger.info("closeChannel: channel already closed");
            return;
        }
        SocketAddress addrRemote = channel.remoteAddress();
        if (this.channelTables.containsKey(addr)) {
            ChannelFuture channelFuture = this.channelTables.get(addr);
            if (channelFuture.channel() == channel) {
                this.channelTables.remove(addr);
            }
        }
        channel.close().addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                logger.info("closeChannel: close the connection to remote address[{}] result: {}", addrRemote,
                        future.isSuccess());
            }
        });
    }
    //关闭一个指定的连接
    public void closeChannel(Channel channel){
        for (Map.Entry<String, ChannelFuture> entry : this.channelTables.entrySet()) {
            if(entry.getValue() == channel){
                this.channelTables.remove(entry.getKey(),entry.getValue());
                break;
            }
        }
        SocketAddress remoteAddress = channel.remoteAddress();
        channel.close().addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                logger.info("closeChannel: close the connection to remote address[{}] result: {}", remoteAddress,
                        future.isSuccess());
            }
        });
    }
    //快速失败，触发channel上所有待处理的ResponseFuture
    public void fastfail(Channel channel){
        for (Map.Entry<Integer, ResponseFuture> entry : responseTable.entrySet()) {
            if (entry.getValue().getProcessChannel() == channel) {
                ResponseFuture responseFuture = responseTable.get(entry.getKey());
                if(responseFuture != null){
                    responseFuture.setSendMessageOk(false);
                    responseFuture.putResponse(null);
                    try{
                        responseFuture.executeCallbackOnce();
                    }catch (Exception e){
                        logger.warn("execute callback in requestFail, and callback throw", e);
                    }
                }
            }
        }
    }

    public void updateNameServerAddressList(List<String> addrs){
        List<String> old = this.namesrvAddrList.get();
        boolean update = false;

        if (!addrs.isEmpty()) {
            if (null == old) {
                update = true;
            } else if (addrs.size() != old.size()) {
                update = true;
            } else {
                for (int i = 0; i < addrs.size() && !update; i++) {
                    if (!old.contains(addrs.get(i))) {
                        update = true;
                    }
                }
            }

            if (update) {
                Collections.shuffle(addrs);
                logger.info("name server address updated. NEW : {} , OLD: {}", addrs, old);
                this.namesrvAddrList.set(addrs);

                if (!addrs.contains(this.namesrvAddrChoosed.get())) {
                    this.namesrvAddrChoosed.set(null);
                }
            }
        }
    }
}
